/*
 * Copyright (C) 2022  即时通讯网(52im.net) & Jack Jiang.
 * The MobileIMSDK v6.1 Project.
 * All rights reserved.
 *
 * > Github地址：https://github.com/JackJiang2011/MobileIMSDK
 * > 文档地址：  http://www.52im.net/forum-89-1.html
 * > 技术社区：  http://www.52im.net/
 * > 技术交流群：320837163 (http://www.52im.net/topic-qqgroup.html)
 * > 作者公众号：“【即时通讯技术圈】”，欢迎关注！
 * > 联系作者：  http://www.52im.net/thread-2792-1-1.html
 *
 * "即时通讯网(52im.net) - 即时通讯开发者社区!" 推荐开源工程。
 *
 * QoS4ReciveDaemonRoot.java at 2022-7-12 16:35:57, code by Jack Jiang.
 */
package com.nami.logic.qos;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.*;

/**
 * 52im qos 主要维护一个定时任务，定时删除过期key 防止客户端发送重复消息
 * @author jack jiang
 *
 */
@Slf4j
public class QosReceiveDaemonRoot {
    private boolean DEBUG = false;
    private int CHECH_INTERVAL = 5 * 60 * 1000; // 5分钟
    private int MESSAGES_VALID_TIME = 10 * 60 * 1000; // 10分钟
    private ConcurrentMap<String, Long> receivedMessages = new ConcurrentHashMap<String, Long>();
    private ScheduledExecutorService scheduledExecutorService = null;
    private Runnable runnable = null;
    private boolean _excuting = false;
    private String debugTag = "";

    public QosReceiveDaemonRoot(int CHECH_INTERVAL, int MESSAGES_VALID_TIME
        , boolean DEBUG, String debugTag) {
        if (CHECH_INTERVAL > 0) {
            this.CHECH_INTERVAL = CHECH_INTERVAL;
        }
        if (MESSAGES_VALID_TIME > 0) {
            this.MESSAGES_VALID_TIME = MESSAGES_VALID_TIME;
        }
        this.DEBUG = DEBUG;
        this.debugTag = debugTag;
    }

    private void doTaskOnce() {
        if (!_excuting) {
            _excuting = true;

            if (DEBUG) {
                log.debug("【NAMI" + this.debugTag + "】【QoS接收方】+++++ START 暂存处理线程正在运行中，当前长度" + receivedMessages.size() + ".");
            }
            //** 遍历HashMap方法二（在大数据量情况下，方法二的性能要5倍优于方法一）
            Iterator<Entry<String, Long>> entryIt = receivedMessages.entrySet().iterator();
            while (entryIt.hasNext()) {
                Entry<String, Long> entry = entryIt.next();
                String key = entry.getKey();
                long value = entry.getValue();

                long delta = System.currentTimeMillis() - value;
                if (delta >= MESSAGES_VALID_TIME) {
                    if (DEBUG) {
                        log.debug("【NAMI" + this.debugTag + "】【QoS接收方】指纹为" + key + "的包已生存" + delta
                            + "ms(最大允许" + MESSAGES_VALID_TIME + "ms), 马上将删除之.");
                    }
                    receivedMessages.remove(key);
                }
            }
        }

        if (DEBUG) {
            log.debug("【NAMI" + this.debugTag + "】【QoS接收方】+++++ END 暂存处理线程正在运行中，当前长度" + receivedMessages.size() + ".");
        }
        //
        _excuting = false;
    }

    public void startup() {
        stop();
        if (receivedMessages != null && receivedMessages.size() > 0) {
            for (String key : receivedMessages.keySet()) {
                putImpl(key);
            }
        }
        scheduledExecutorService = new ScheduledThreadPoolExecutor(1
            , ThreadFactoryBuilder.create()
            .setNamePrefix("nami-id-expire-task")
            .setDaemon(true)
            .setUncaughtExceptionHandler(((thread, throwable) ->
                log.error("[nami-id-expire-task] ThreadPool {} got exception", thread, throwable)))
            .build());
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                doTaskOnce();
            }
        }, CHECH_INTERVAL, CHECH_INTERVAL, TimeUnit.MILLISECONDS);

        log.debug("【NAMI" + this.debugTag + "】【QoS发送方】====== 校验客户端是否重复发送已成功启动");

    }

    public void stop() {
        if (scheduledExecutorService != null) {
            try {
                scheduledExecutorService.shutdown();
            } finally {
                scheduledExecutorService = null;
            }
        }
    }

    public boolean isRunning() {
        return scheduledExecutorService != null;
    }

    public void addReceived(String fingerPrintOfClientMsgId) {
        if (fingerPrintOfClientMsgId == null) {
            log.debug("【NAMI" + this.debugTag + "】无效的 fingerPrintOfProtocal==null!");
            return;
        }

        if (receivedMessages.containsKey(fingerPrintOfClientMsgId)) {
            log.debug("【NAMI" + this.debugTag + "】【QoS接收方】指纹为" + fingerPrintOfClientMsgId
                + "的消息已经存在于接收列表中，该消息重复了（原理可能是对方因未收到应答包而错误重传导致），更新收到时间戳哦.");
        }

        putImpl(fingerPrintOfClientMsgId);
    }

    private void putImpl(String fingerPrintOfClientMsgId) {
        if (fingerPrintOfClientMsgId != null) {
            receivedMessages.put(fingerPrintOfClientMsgId, System.currentTimeMillis());
        }
    }

    public boolean hasReceived(String fingerPrintOfClientMsgId) {
        return receivedMessages.containsKey(fingerPrintOfClientMsgId);
    }

    public int size() {
        return receivedMessages.size();
    }

    public QosReceiveDaemonRoot setDebugable(boolean debugable) {
        this.DEBUG = debugable;
        return this;
    }

    public boolean isDebugable() {
        return this.DEBUG;
    }
}
